package com.atguigu.realtime.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV1;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.AtguiguUtil;
import com.atguigu.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author lzc
 * @Date 2022/5/20 9:27
 */
public class Dwd_01_BaseLogApp extends BaseAppV1 {
    public static void main(String[] args) {
        new Dwd_01_BaseLogApp().init(3001, 2, "Dwd_01_BaseLogApp", Constant.TOPIC_ODS_LOG);
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          DataStreamSource<String> stream) {
        // 1. etl
        SingleOutputStreamOperator<String> etledStream = etl(stream);
        // 2. 新老客户纠正
        SingleOutputStreamOperator<JSONObject> validatedStream = validateNewOrOld(etledStream);
        // 3. 分流
        Map<String, DataStream<String>> fiveStreams = splitStream(validatedStream);
        
        // 4. 不同的流写入到不同的topic
        writeToKafka(fiveStreams);
    }
    
    private void writeToKafka(Map<String, DataStream<String>> fiveStreams) {
        fiveStreams.get("page").addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_PAGE));
        fiveStreams.get("err").addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ERROR));
        fiveStreams.get("display").addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_DISPLAY));
        fiveStreams.get("start").addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_START));
        fiveStreams.get("action").addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ACTION));
    }
    
    private Map<String, DataStream<String>> splitStream(SingleOutputStreamOperator<JSONObject> stream) {
        /*
        5个流:
            启动 主流
            页面 曝光 错误 行为 测输出流
        
         */
        OutputTag<String> errTag = new OutputTag<String>("err") {};
        OutputTag<String> pageTag = new OutputTag<String>("page") {};
        OutputTag<String> displayTag = new OutputTag<String>("display") {};
        OutputTag<String> actionTag = new OutputTag<String>("action") {};
        
        
        SingleOutputStreamOperator<String> startStream = stream
            .process(new ProcessFunction<JSONObject, String>() {
                @Override
                public void processElement(JSONObject obj,
                                           Context ctx,
                                           Collector<String> out) throws Exception {
                    
                    // 1. 错误日志
                    JSONObject err = obj.getJSONObject("err");
                    if (err != null) {
                        ctx.output(errTag, obj.toJSONString());
                        
                        obj.remove("err");
                    }
                    
                    // 2. 启动
                    JSONObject start = obj.getJSONObject("start");
                    if (start != null) {
                        out.collect(obj.toJSONString());  // 启动日志放入主流
                        
                    } else {
                        // 其他日志
                        JSONObject common = obj.getJSONObject("common");
                        JSONObject page = obj.getJSONObject("page");
                        Long ts = obj.getLong("ts");
                        
                        // 1. 是否为曝光日志
                        JSONArray displays = obj.getJSONArray("displays");
                        if (displays != null) {
                            for (int i = 0; i < displays.size(); i++) {
                                JSONObject display = displays.getJSONObject(i);
                                
                                display.putAll(common);
                                display.putAll(page);
                                display.put("ts", ts);
                                ctx.output(displayTag, display.toJSONString());
                            }
                            obj.remove("displays");
                        }
                        // 2. 是否为行为日志
                        
                        JSONArray actions = obj.getJSONArray("actions");
                        if (actions != null) {
                            for (int i = 0; i < actions.size(); i++) {
                                JSONObject action = actions.getJSONObject(i);
                                
                                action.putAll(common);
                                action.putAll(page);
                                action.put("ts", ts);
                                
                                ctx.output(actionTag, action.toJSONString());
                                
                            }
                            obj.remove("actions");
                        }
                        
                        ctx.output(pageTag, obj.toJSONString());
                        
                    }
                    
                }
            });
    
        DataStream<String> pageStream = startStream.getSideOutput(pageTag);
        DataStream<String> actionStream = startStream.getSideOutput(actionTag);
        DataStream<String> displayStream = startStream.getSideOutput(displayTag);
        DataStream<String> errStream = startStream.getSideOutput(errTag);
    
        Map<String, DataStream<String>> map = new HashMap<String, DataStream<String>>();
        map.put("start", startStream);
        map.put("err", errStream);
        map.put("display", displayStream);
        map.put("action", actionStream);
        map.put("page", pageStream);
        
        return map;
    
    
    }
    
    private SingleOutputStreamOperator<JSONObject> validateNewOrOld(SingleOutputStreamOperator<String> stream) {
        /*
            is_new 0 1
            
            is_new = 1
                状态 null
                    确实是新用户
                 状态 not null
                    是老用户
                
            
            is_new = 0
                不用操作 TODO
         */
        return stream
            .map(JSON::parseObject)
            .keyBy(obj -> obj.getJSONObject("common").getString("mid"))
            .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                
                private ValueState<String> firstVisitDateState;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    firstVisitDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("firstVisitDateState", String.class));
                }
                
                @Override
                public void processElement(JSONObject obj,
                                           Context ctx,
                                           Collector<JSONObject> out) throws Exception {
                    Long ts = obj.getLong("ts");
                    String date = AtguiguUtil.toDate(ts, "yyyy-MM-dd");
                    
                    JSONObject common = obj.getJSONObject("common");
                    String isNew = common.getString("is_new");
                    
                    
                    String firstVisitDate = firstVisitDateState.value();
                    
                    if ("1".equals(isNew)) {
                        // 可能会有误判, 通过状态纠正
                        if (firstVisitDate == null) {
                            // 证明以前确实没有访问过
                            firstVisitDateState.update(date);
                        } else {
                            if (!date.equals(firstVisitDate)) {  // 当天和状态的日志是否一致
                                // is_new是1, 但是状态有, 并且又不是第一天访问, 表示你清除过缓存或者重装过app
                                common.put("is_new", "0");
                            }
                        }
                    } else if (firstVisitDate == null) {
                        // 今天的是0, 不会维护状态.
                        // 明天这个用户重装app, 会来 1
                        // 把状态维护一个以前的日期
                        firstVisitDateState.update(AtguiguUtil.toDate(ts - 24 * 60 * 60 * 1000));
                    }
                    
                    out.collect(obj);
                    
                }
            });
        
        
    }
    
    private SingleOutputStreamOperator<String> etl(DataStreamSource<String> stream) {
        return stream.filter(json -> {
            try {
                JSON.parseObject(json);
                return true;
            } catch (Exception e) {
                System.out.println("json格式有误, 请检查....");
                return false;
            }
        });
    }
}
